/*
 * UFixedPacketQueue.cpp
 *
 *  Created on: 2021年4月15日
 *      Author: chuan
 */

#include "UFixedPacketQueue.h"
#include <deque>
#include <map>
#include "TCriticalSection.h"
#include "TEvent.h"
#include "ClockTime.h"


typedef std::pair<int64_t, UMediaPacket>    ClockMediaPacket;

typedef std::deque< ClockMediaPacket >  PacketDeque;


class UFixedPacketQueuePrivate
{
public:
    PacketDeque m_packets;
    comn::CriticalSection   m_cs;
    comn::Event m_event;
	int	m_maxDuration;
	bool m_eventEnabled;

	UFixedPacketQueuePrivate():
		m_maxDuration(),
		m_eventEnabled(true)
	{
	}

    size_t findVideoKey()
    {
        for (size_t i = 0; i < m_packets.size(); i ++)
        {
            UMediaPacket& pkt = m_packets[i].second;
            if (pkt.isVideo() && pkt.isKey())
            {
                return i;
            }
        }
        return -1;
    }

    size_t rfindVideoKey()
    {
        if (m_packets.size() < 2)
        {
            return -1;
        }

        for (size_t i = m_packets.size() - 1; i != 0; i --)
        {
            UMediaPacket& pkt = m_packets[i].second;
            if (pkt.isVideo() && pkt.isKey())
            {
                return i;
            }
        }
        return -1;
    }


    size_t size()
    {
        comn::AutoCritSec   lock(m_cs);
        return m_packets.size();
    }

    bool empty()
    {
        comn::AutoCritSec   lock(m_cs);
        return m_packets.empty();
    }

    size_t push(UMediaPacket& packet)
    {
        size_t count = 0;
        {
            comn::AutoCritSec   lock(m_cs);
            m_packets.push_back(ClockMediaPacket(comn::ClockTime::getTime(), packet));

            count = m_packets.size();

			if (m_maxDuration > 0)
			{
				while (m_packets.size() > 1)
				{
					int duration = (int)countClockDuration();
					if (duration > m_maxDuration)
					{
						m_packets.pop_front();
					}
					else
					{
						break;
					}
				}
			}
        }

        if (m_eventEnabled)
        {
            m_event.post();
        }

        return count;
    }

    UMediaPacket pop()
    {
        UMediaPacket packet;
        pop(packet);
        return packet;
    }

    bool pop(UMediaPacket& packet)
    {
        bool found = false;
        comn::AutoCritSec   lock(m_cs);
        if (!m_packets.empty())
        {
            packet = m_packets.front().second;
            m_packets.pop_front();
            found = true;
        }
        return found;
    }

    void clear()
    {
        {
            comn::AutoCritSec   lock(m_cs);
            m_packets.clear();
        }

        m_event.post();
    }

    bool timedwait(int ms)
    {
        return m_event.timedwait(ms);
    }

    bool pop(UMediaPacket& packet, int ms)
    {
        if (empty())
        {
            timedwait(ms);
        }
        return pop(packet);
    }

    void cancelWait()
    {
        m_event.post();
    }

    int64_t getDuration()
    {
        comn::AutoCritSec   lock(m_cs);

        if (m_packets.size() < 2)
        {
            return 0;
        }

        return m_packets.back().second.pts() - m_packets.front().second.pts();
    }

    size_t dropUntilKeyFrame()
    {
        comn::AutoCritSec   lock(m_cs);

        size_t count = 0;
        size_t toIdx = rfindVideoKey();
        if (toIdx != (size_t)-1)
        {
            PacketDeque::iterator itBegin = m_packets.begin();
            PacketDeque::iterator itEnd = itBegin;
            std::advance(itEnd, toIdx);
            m_packets.erase(itBegin, itEnd);
            count = toIdx;
        }
        return count;
    }

    int64_t getClockDuration()
    {
		comn::AutoCritSec   lock(m_cs);
		return countClockDuration();
    }

    int64_t reserve(int64_t duration)
    {
		int64_t ts = comn::ClockTime::getTime() - duration;
		comn::AutoCritSec   lock(m_cs);

		PacketDeque::iterator it = m_packets.begin();
		while (it != m_packets.end())
		{
			ClockMediaPacket& pkt = *it;
			if (pkt.second.isVideoKey() && pkt.first >= ts)
			{
				break;
			}
			else
			{
				it = m_packets.erase(it);
			}
		}

		return countClockDuration();
    }

    size_t readAll(UFixedPacketQueue::PacketHandler handler)
    {
		size_t count = 0;
		comn::AutoCritSec   lock(m_cs);
		for (ClockMediaPacket& pkt : m_packets)
		{
			handler(pkt.second);
			count++;
		}
		return count;
    }

	int64_t countClockDuration()
	{
		if (m_packets.size() < 2)
		{
			return 0;
		}

		return m_packets.back().first - m_packets.front().first;
	}

	int setMaxClockDuration(int ms)
	{
		comn::AutoCritSec   lock(m_cs);
		int maxDuration = m_maxDuration;
		m_maxDuration = ms;
		return maxDuration;
	}

	void enableEvent(bool enabled)
	{
	    m_eventEnabled = enabled;
	}

};


UFixedPacketQueue::UFixedPacketQueue():
        m_pimp(new UFixedPacketQueuePrivate())
{
}

UFixedPacketQueue::~UFixedPacketQueue()
{
    delete m_pimp;
}


size_t UFixedPacketQueue::size()
{
    return m_pimp->size();
}

bool UFixedPacketQueue::empty()
{
    return m_pimp->empty();
}

size_t UFixedPacketQueue::push(UMediaPacket& packet)
{
    return m_pimp->push(packet);
}

UMediaPacket UFixedPacketQueue::pop()
{
    return m_pimp->pop();
}

bool UFixedPacketQueue::pop(UMediaPacket& packet)
{
    return m_pimp->pop(packet);
}

void UFixedPacketQueue::clear()
{
    m_pimp->clear();
}

bool UFixedPacketQueue::timedwait(int ms)
{
    return m_pimp->timedwait(ms);
}

bool UFixedPacketQueue::pop(UMediaPacket& packet, int ms)
{
    if (empty())
    {
        timedwait(ms);
    }
    return pop(packet);
}

void UFixedPacketQueue::cancelWait()
{
    m_pimp->cancelWait();
}

int64_t UFixedPacketQueue::getDuration()
{
    return m_pimp->getDuration();
}

size_t UFixedPacketQueue::dropUntilKeyFrame()
{
    return m_pimp->dropUntilKeyFrame();
}

int UFixedPacketQueue::setMaxClockDuration(int ms)
{
	return m_pimp->setMaxClockDuration(ms);
}

int64_t UFixedPacketQueue::getClockDuration()
{
    return m_pimp->getClockDuration();
}

int64_t UFixedPacketQueue::reserve(int64_t duration)
{
    return m_pimp->reserve(duration);
}

size_t UFixedPacketQueue::readAll(PacketHandler handler)
{
    return m_pimp->readAll(handler);
}

void UFixedPacketQueue::enableEvent(bool enabled)
{
    m_pimp->enableEvent(enabled);
}
